package com.ithinks.flow.common.utils.middleware.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.io.IOException;

/**
 * @Description
 * @ClassName kafkaConsumer
 * @Author MR.LI
 * @date 2020.11.25 17:38
 */
@Component
public class kafkaConsumer {

    @Value("${kafkaConfig.topic}")
    private String topic;

    private static final Logger logger = LoggerFactory.getLogger(kafkaConsumer.class);

    /**
     * 实时获取kafka数据(生产一条，监听生产topic自动消费一条)	 * @param record	 * @throws IOException
     */
    @KafkaListener(topics = {"${kafkaConfig.topic}"})
    public void listen(ConsumerRecord<?, ?> record) throws IOException {
        String value = (String) record.value();
        System.out.println("收到的消息："+value);
    }


//######kafka配置信息######
//    kafkaConfig:
//    address: 127.0.0.1:9002,127.0.0.2:9003
//            #(是否自动提交)
//    autoCommit: true
//            #(连接超时时间)
//    timeout: 20000
//    commitInterval: 100
//            #(消费组)
//    groupId: ivrKafKa
//   #(实时生产，实时消费，不会从头开始消费)
//    offsetReset: latest
//  #(设置消费线程数)
//    concurrency: 15
//            #消费的topic
//    topic: ivrFlowUpdate
//    batchSize: 4096
//    linger: 1

}
